package cn.net.sinodata.job.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.net.sinodata.common.core.constant.CacheNames;
import cn.net.sinodata.common.core.constant.TransChannelConstants;
import cn.net.sinodata.common.core.enums.DelFlagEnum;
import cn.net.sinodata.common.core.enums.LockKeyEnum;
import cn.net.sinodata.common.core.enums.MsgParseHandlerTypeEnum;
import cn.net.sinodata.common.core.enums.MsgParseProcessStateEnum;
import cn.net.sinodata.common.hazelcast.utils.HazelcastUtils;
import cn.net.sinodata.system.domain.BmTransLogMsg;
import cn.net.sinodata.system.domain.BmTransLogRecord;
import cn.net.sinodata.system.domain.SysDictData;
import cn.net.sinodata.system.mapper.BmTransLogMsgMapper;
import cn.net.sinodata.system.mapper.BmTransLogRecordMapper;
import cn.net.sinodata.system.msghandler.process.*;
import cn.net.sinodata.system.msghandler.process.base.IMsgProcessHandler;
import cn.net.sinodata.system.service.IBmDistributedLockService;
import cn.net.sinodata.threadpool.base.AbstractThreadPool;
import cn.net.sinodata.threadpool.base.QueueThreadPoolManager;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@Slf4j
@Component
public class MsgParseJobService {

    // 使用到的线程池的名称
    private static final String MSG_PARSE_THREAD_POOL = "MsgParseThreadPool";
    // 使用到的字典名
    private static final String SINGLE_MSG_PARSE_GET_NUM_DICT_TYPE = "single_msg_parse_get_num";

    @Autowired
    private BmTransLogRecordMapper bmTransLogRecordMapper;
    @Autowired
    private BmTransLogMsgMapper bmTransLogMsgMapper;
    @Autowired
    private IBmDistributedLockService lock;

    /**
     * 解析报文入口定时任务
     */
    @XxlJob("msgParseJobHandler")
    public void msgParse() {
        log.info("获取待解析报文定时任务-开始");
        // 获取对应的线程池
        AbstractThreadPool threadPool = QueueThreadPoolManager.getInstance().getThreadPool(MSG_PARSE_THREAD_POOL);
        if (Objects.isNull(threadPool)) {
            log.warn("未获取到解析报文线程池处理类");
            XxlJobHelper.handleFail("未获取到解析报文线程池处理类");
            return;
        }

        // 从缓存中拿到本地需要查询的报文记录数
        int singleMsgParseGetNum = 100;
        List<SysDictData> sysDictDatas = HazelcastUtils.getCache(CacheNames.SYS_DICT_DATA, SINGLE_MSG_PARSE_GET_NUM_DICT_TYPE);
        if (CollectionUtil.isNotEmpty(sysDictDatas)) {
            String dictValue = sysDictDatas.get(0).getDictValue();
            singleMsgParseGetNum = Integer.valueOf(dictValue);
        }

        // 查询待处理任务
        List<BmTransLogRecord> bmTransLogRecords = null;
        try {
            if (!lock.lockWithWait(LockKeyEnum.GET_MSG_RECORD_LOCK, 3000)) {
                log.warn("获取锁失败");
                return;
            }

            LambdaQueryWrapper<BmTransLogRecord> lqw = new LambdaQueryWrapper<BmTransLogRecord>()
                .eq(BmTransLogRecord::getProcessState, MsgParseProcessStateEnum.PENDING.code)
                .eq(BmTransLogRecord::getDelFlag, DelFlagEnum.EXISTS.code)
                .orderByAsc(BmTransLogRecord::getCreateTime);

            bmTransLogRecords = bmTransLogRecordMapper.selectList(new Page<>(1, singleMsgParseGetNum), lqw);
            if (CollectionUtil.isEmpty(bmTransLogRecords)) {
                log.info("未获取到待处理的报文记录");
                return;
            }
            log.info("获取待处理的报文记录数为：{}", bmTransLogRecords.size());
            // 更新任务状态为处理中
            List<BmTransLogRecord> collect = bmTransLogRecords.stream().map(entry -> {
                entry.setProcessState(MsgParseProcessStateEnum.PROCESSING.code);
                return entry;
            }).collect(Collectors.toList());
            bmTransLogRecordMapper.updateBatchById(collect);
        } catch (Exception e) {
            // TODO 线程被中断，这是是否需要特殊处理
            log.error("获取锁被中断:", e);
        } finally {
            lock.unlock(LockKeyEnum.GET_MSG_RECORD_LOCK);
        }

        // 创建线程池任务
        for (BmTransLogRecord bmTransLogRecord : bmTransLogRecords) {
            String id = bmTransLogRecord.getId();
            BmTransLogMsg bmTransLogMsg = bmTransLogMsgMapper.selectOne(new LambdaQueryWrapper<BmTransLogMsg>().eq(BmTransLogMsg::getTransLogRecordId, id));
            // 判断渠道
            String transChannelCode = bmTransLogRecord.getTransChannelCode();
            IMsgProcessHandler msgHandler = switch (transChannelCode){
                case TransChannelConstants.CFETS -> new CFETSMsgProcessHandler(bmTransLogRecord, bmTransLogMsg, MsgParseHandlerTypeEnum.XML);
                case TransChannelConstants.CCDC -> new CCDCMsgProcessHandler(bmTransLogRecord, bmTransLogMsg, MsgParseHandlerTypeEnum.XML);
                case TransChannelConstants.SHCH -> new SHCHMsgProcessHandler(bmTransLogRecord, bmTransLogMsg, MsgParseHandlerTypeEnum.XML);
                case TransChannelConstants.STP -> new STPMsgProcessHandler(bmTransLogRecord, bmTransLogMsg, MsgParseHandlerTypeEnum.XML);
                case TransChannelConstants.IPS -> new IPSMsgProcessHandler(bmTransLogRecord, bmTransLogMsg, MsgParseHandlerTypeEnum.XML);
                default -> null;
            };

            if (!Objects.isNull(msgHandler)) {
                threadPool.execute(msgHandler);
            } else {
                log.warn(transChannelCode + "渠道类型不支持");
                bmTransLogRecord.setProcessState(MsgParseProcessStateEnum.NEEDLESS.code);
                bmTransLogRecord.setStateComment("渠道类型["+transChannelCode+"]不支持");
                bmTransLogRecordMapper.updateById(bmTransLogRecord);
            }
        }
        log.info("获取待解析报文定时任务-结束");
    }

}
